package hbase

import (
	"github.com/tsuna/gohbase/hrpc"
	"fmt"
	"EasyData/global"
	log "code.google.com/p/log4go"
	"context"
)
/**
	将mysql的数据存储到Hbase中
 */
func NewRowsInToHbase(item map[string]interface{},task *global.Task) error{
	values,table,key,err:=sqlToBase(item,task)
	if err != nil {
		errStr := fmt.Sprintf("taskName [%s] taskCommand [%s] converDataFromMysqlToHabase error table[%s] row[%s] values[%s] error [%s]",
			task.TaskName,task.TaskCommond,table,key,values,err)
		log.Error(errStr)
		return err
	}
	_,err = newRows(values,table,key)
	if err != nil {
		errStr := fmt.Sprintf("taskName [%s] taskCommand [%s] insert into hbase table[%s] row[%s] values[%s] error [%s]",
			task.TaskName,task.TaskCommond,table,key,values,err)
		log.Error(errStr)
		return err
	}
	log.Info("taskName [%s] taskCommand [%s] insert into hbase table[%s] row[%s] values[%s] success",
		task.TaskName,task.TaskCommond,table,key,values)
	return nil
}

/**
	将数据存储到Hbase中
 */
func newRows(values map[string]map[string][]byte,table string,key string) (*hrpc.Result,error){
	putRequest,errs := hrpc.NewPutStr(context.Background(), table, key, values)
	rsp, errs := client.Put(putRequest)
	return rsp,errs
}
